package net.lab1024.sa.admin.module.open.heartbeat;

import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import net.lab1024.sa.admin.module.device.terminal.DeviceTerminalService;
import net.lab1024.sa.admin.module.device.terminal.domain.DeviceTerminalEntity;
import net.lab1024.sa.admin.module.open.heartbeat.domain.DeviceHeartBeatInfo;
import net.lab1024.sa.admin.module.open.heartbeat.domain.DeviceHeartbeatReportForm;
import net.lab1024.sa.base.common.util.SmartIpUtil;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;

@Slf4j
@Service
public class DeviceHeartbeatService {

    private final DeviceTerminalService deviceTerminalService;

    public DeviceHeartbeatService(DeviceTerminalService deviceTerminalService) {
        this.deviceTerminalService = deviceTerminalService;
    }

    // 阻塞队列，用于心跳请求过来先放入阻塞队列里
    // 当然，如果害怕重启项目丢失心跳数据，可以使用Redis中的Streams来实现阻塞队列的效果
    // 使用Redis的前提是需要持久化所有的心跳数据
    // 目前只是记录最新的心跳数据，所有可以解释重启时心跳数据丢失的情况，因为下次心跳还好再上报
    private final BlockingQueue<DeviceHeartBeatInfo> heartbeatQueue = new LinkedBlockingQueue<>();

    private final static Object RUNNING_LOCK = new Object();

    // 处理心跳信息的运行状态
    private final AtomicBoolean running = new AtomicBoolean(true);

    // 单个线程已处理心跳记录
    private final AtomicInteger processed = new AtomicInteger(0);

    // 缓存心跳信息列表
    private final List<DeviceHeartBeatInfo> heartbeatInfos = new ArrayList<>();

    // 缓存心跳信息列表最大长度
    private final static int MAX_SIZE = 200;

    /**
     * 将值放入队列
     */
    public void putHeartbeatInQueue(DeviceHeartbeatReportForm form, String ip) {
        // 生成bo对象
        DeviceHeartBeatInfo deviceHeartBeatInfo = DeviceHeartBeatInfo.builder()
                .serialno(form.getSerialno())
                .osType(form.getOsType())
                .heartbeatTime(form.getHeartbeatTime())
                .ip(ip)
                .cpuCores(form.getCpuCores())
                .memorySize(form.getMemorySize())
                .diskSize(form.getDiskSize())
                .build();
        try {
            heartbeatQueue.put(deviceHeartBeatInfo);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 标记线程中断状态
            log.error("将心跳信息添加到队列时出错", e);
            throw new RuntimeException(e);
        }

    }

    /**
     * 获取心跳队列长度
     */
    public Integer getHeartbeatQueueSize() {
        return heartbeatQueue.size();
    }

    @Async
    public CompletableFuture<String> heartbeatProcessing() {
        synchronized (RUNNING_LOCK) {
            if (!running.get()) {
                return CompletableFuture.completedFuture("心跳已有线程在处理");
            }
            running.set(false);
        }
        while (!heartbeatQueue.isEmpty() || !heartbeatInfos.isEmpty()) {
            DeviceHeartBeatInfo form = heartbeatQueue.poll(); // 从队列中取出请求参数
            // 存到list中
            if (form != null) {
                heartbeatInfos.add(form);
            }
            // 集合元素超过MAX_SIZE，或者最后一次取队列的值为null时，将数据批量存入数据库
            if (!heartbeatInfos.isEmpty() && (heartbeatInfos.size() >= MAX_SIZE || form == null)) {
                saveHeartbeatInfo();
            }
        }
        running.set(true);

        //
        String threadName = Thread.currentThread().getName();
        int processedCount = processed.getAndSet(0);
        String processInfo = StrUtil.format("本次线程 {} 处理心跳 {} 条", threadName, processedCount);
        return CompletableFuture.completedFuture(processInfo);
    }

    /**
     * 检查心跳
     *
     * @return 心跳丢失信息
     */
    @Async
    public CompletableFuture<String> checkHeartbeat() {
        LocalDateTime fiveMinutesAgo = LocalDateTime.now().minusMinutes(5);
        return CompletableFuture.completedFuture(deviceTerminalService.lostByTimeAgo(fiveMinutesAgo));
    }

    /**
     * 保存心跳信息
     */
    private void saveHeartbeatInfo() {
        ArrayList<DeviceTerminalEntity> deviceTerminalEntityList = new ArrayList<>();
        for (DeviceHeartBeatInfo heartbeatInfo : heartbeatInfos) {
            DeviceTerminalEntity entity = new DeviceTerminalEntity();
            entity.setSerialno(heartbeatInfo.getSerialno());
            entity.setOsType(heartbeatInfo.getOsType());
            entity.setHeartbeatTime(heartbeatInfo.getHeartbeatTime());
            entity.setLinkFlag(Boolean.TRUE);
            entity.setIp(heartbeatInfo.getIp());
            entity.setCpuCores(heartbeatInfo.getCpuCores());
            entity.setMemorySize(heartbeatInfo.getMemorySize());
            entity.setDiskSize(heartbeatInfo.getDiskSize());
            try {
                // 解析ip
                List<String> regionList = SmartIpUtil.getRegionList(heartbeatInfo.getIp());
                setRegionInfo(entity, regionList.get(0), DeviceTerminalEntity::setCountry);
                setRegionInfo(entity, regionList.get(2), DeviceTerminalEntity::setProvince);
                setRegionInfo(entity, regionList.get(3), DeviceTerminalEntity::setCity);
                setRegionInfo(entity, regionList.get(4), DeviceTerminalEntity::setIsp);
            } catch (RuntimeException e) {
                log.error("解析ip信息时出错", e);
            }
            deviceTerminalEntityList.add(entity);
        }
        processed.addAndGet(heartbeatInfos.size());
        heartbeatInfos.clear();
        // 保存设备信息
        deviceTerminalService.saveDeviceTerminalBatch(deviceTerminalEntityList);
    }

    /** 将解析后的ip信息set到实体 */
    private void setRegionInfo(DeviceTerminalEntity entity,
                               String value,
                               BiConsumer<DeviceTerminalEntity, String> setter) {
        if (!"0".equals(value)) {
            setter.accept(entity, value);
        }
    }

}
